package cn.piesat.scanning.business.dq1043.config;

import cn.piesat.scanning.utils.ConfigPropertiesUtil;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Properties;

/**
 * Kafka消费者工厂配置
 *
 * @author xhl
 * @Date 2020/8/16
 */
public class KafkaConsumerFactory {

    static KafkaConsumer<String, String> consumer;
    public static KafkaConsumer getKafkaConsumer(String group) {
        Properties props = new Properties();
        props.put("bootstrap.servers", ConfigPropertiesUtil.get("application","spring.kafka.bootstrap-servers"));
        //每个消费者分配独立的组号
        props.put("group.id", group);
        //如果value合法，则自动提交偏移量
//        props.put("enable.auto.commit", ConfigPropertiesUtil.get("application","spring.kafka.consumer.enable-auto-commit"));
        // 每次拉取10条
        props.put("max.poll.records", ConfigPropertiesUtil.get("application","spring.kafka.consumer.max-poll-records"));
        //设置多久一次更新被消费消息的偏移量
//        props.put("auto.commit.interval.ms", ConfigPropertiesUtil.get("application","spring.kafka.consumer.auto-commit-interval"));
        //设置会话响应的时间，超过这个时间kafka可以选择放弃消费或者消费下一条消息
        props.put("session.timeout.ms", "30000");
        //自动重置offset
        props.put("auto.offset.reset", ConfigPropertiesUtil.get("application","spring.kafka.consumer.auto-offset-reset"));
        props.put("key.deserializer",ConfigPropertiesUtil.get("application","spring.kafka.consumer.key-deserializer"));
        props.put("value.deserializer",ConfigPropertiesUtil.get("application","spring.kafka.consumer.value-deserializer"));

//        props.put("max.partition.fetch.bytes",Integer.MAX_VALUE);
//        props.put("fetch.max.wait.ms","10000000");
        return consumer = new KafkaConsumer<>(props);
    }

}
